package com.jun.common.redis.config;

import com.jun.common.redis.properties.RedisTemplateProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStreamCommands;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author yuejun.li
 * @date 2024/7/15 12:37:24
 */
@Slf4j
@Configuration
@EnableConfigurationProperties({RedisTemplateProperties.class})
public class RedisConfig {

    @PostConstruct
    public void init() {
        log.info("Redis successful!");
    }

    /**
     * 配置redis模板和序列化
     * @param factory
     * @param properties
     * @return
     */
    @Bean
    @Primary
    public RedisTemplate redisTemplate(RedisConnectionFactory factory, RedisTemplateProperties properties) {
        RedisTemplate redisTemplate = new RedisTemplate();
        redisTemplate.setConnectionFactory(factory);
        properties.serialization(redisTemplate);
        return redisTemplate;
    }

    /**
     * 配置发布订阅的关系
     * @param factory
     * @return
     */
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory factory) {
        final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        return container;
    }


    /**
     * 配置redis消息队列
     * @param redisConnectionFactory
     * @param dealThreadExector
     * @return
     */
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(RedisConnectionFactory redisConnectionFactory,
                                                                         @Qualifier("dealThreadExector") ThreadPoolTaskExecutor dealThreadExector){
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                    .builder()
                    //一次最多获取多少条消息
                    .batchSize(3)
                    .executor(dealThreadExector)
                    //Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                    .pollTimeout(Duration.ofSeconds(3))
                    // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                    .errorHandler(ex -> {
                        log.error("异常信息为：{}", ex.getMessage());
                    })
                    .build();
        return StreamMessageListenerContainer.create(redisConnectionFactory, options);
    }
}
